package com.supertool.tong.merger.util;

import java.io.DataOutputStream;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.io.SequenceFile.CompressionType;
import org.apache.hadoop.io.compress.DefaultCodec;

//import com.supertool.admonitor.LogKey;
//import com.supertool.admonitor.LogParser;
//import com.supertool.mobilemonitor.LogKey;
//import com.supertool.mobilemonitor.LogParser;
import com.supertool.tong.merger.util.MzSequenceFile.Writer;

public class MZSequenceOutputFormat<K,V> extends MergeOutputFormat<K,V>{

    @Override
    public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job)
            throws IOException, InterruptedException {

        Configuration conf = job.getConfiguration();
        boolean isCompressed = getCompressOutput(job);
        CompressionCodec codec = null;
        String extension = "";
        if (isCompressed) {
            Class<? extends CompressionCodec> codecClass =
                    getOutputCompressorClass(job, GzipCodec.class);
            codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, conf);
            extension = codec.getDefaultExtension();
        }
        Path file = getDefaultWorkFile(job, extension);
        FileSystem fs = file.getFileSystem(conf);

        final MzSequenceFile.Writer out =
                new MzSequenceFile.Writer(fs, job.getConfiguration(), file, job.getOutputValueClass(), codec);

        return new RecordWriter<K, V>() {
            public void write(K key, V value)
                    throws IOException {
                out.append(key.toString(),value);
            }
            @Override
            public void close(TaskAttemptContext context) throws IOException,
                    InterruptedException {
                out.close();
            }
        };
    }
}
